package com.jvxb.demo.service.impl;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.jvxb.demo.constant.KafkaTopicConstant;
import com.jvxb.demo.entity.NginxAccessLog;
import com.jvxb.demo.service.NginxAccessLogService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.Locale;

/**
 * @author jvxb
 * @since 2023-01-17
 */
@Slf4j
@Service
public class NginxAccessLogServiceImpl implements NginxAccessLogService {

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    @KafkaListener(topics = {KafkaTopicConstant.NGINX_LOG})
    public void handlerNginxLog(ConsumerRecord<String, String> consumerRecord) {
        System.out.println("消费nginx_log: " + consumerRecord.value());
        JSONObject data = JSONObject.parseObject(consumerRecord.value());
        String path = data.getJSONObject("log").getJSONObject("file").getString("path");
        if (path.endsWith("access.log")) {
            String message = data.getString("message");
            NginxAccessLog accessLog = resolveAccessLog(message);
            if (accessLog == null) {
                return;
            }
            accessLog.setNginx_ip(data.getString("intranet_ip"));
            saveAccessLog(accessLog);
        }
    }

    @Override
    public NginxAccessLog resolveAccessLog(String message) {
        //120.xx.xx.xx - - [17/JAN/2023:22:56:30 +0800] \"GET /test HTTP/1.1\" 404 555 \"-\" \"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36\" \"-\"
        String accessLogReg = "(\\S+) - (\\S+) \\[(.+)\\] \"([^\"]+)\" (\\d+) (\\d+) \"([^\"]+)\" \"([^\"]+)\" \"([^\"]+)\"$";
        String remoteAddr = message.replaceFirst(accessLogReg, "$1");
        if (remoteAddr == null || remoteAddr.length() > 16) {
            log.error("解析nginx access_log文本异常，文本内容：{}", message);
            return null;
        }
        try {
            NginxAccessLog accessLog = new NginxAccessLog();
            accessLog.setRemote_addr(remoteAddr);
            accessLog.setRemote_user(message.replaceFirst(accessLogReg, "$2"));
            accessLog.setTime_local(message.replaceFirst(accessLogReg, "$3"));
            accessLog.setTime(DateUtil.parse(accessLog.getTime_local(), "dd/MMM/yyyy:HH:mm:ss +0800", Locale.ENGLISH).getTime());
            accessLog.setTime_local(DateUtil.format(DateUtil.date(accessLog.getTime()), DatePattern.NORM_DATETIME_PATTERN));
            accessLog.setRequest(message.replaceFirst(accessLogReg, "$4"));
            accessLog.setRequest_url(accessLog.getRequest().contains(" ") ? accessLog.getRequest().split(" ")[1] : null);
            accessLog.setStatus(Integer.valueOf(message.replaceFirst(accessLogReg, "$5")));
            accessLog.setBody_bytes_sent(Integer.valueOf(message.replaceFirst(accessLogReg, "$6")));
            accessLog.setHttp_referer(message.replaceFirst(accessLogReg, "$7"));
            accessLog.setHttp_user_agent(message.replaceFirst(accessLogReg, "$8"));
            accessLog.setHttp_x_forwarded_for(message.replaceFirst(accessLogReg, "$9"));
            return accessLog;
        } catch (Exception e) {
            log.error("解析nginx access_log文本异常，文本内容：{}", message, e);
        }
        return null;
    }

    @Override
    public void saveAccessLog(NginxAccessLog nginxAccessLog) {
        if (nginxAccessLog == null) {
            return;
        }
        try {
            IndexRequest indexRequest = new IndexRequest()
                    .index(NginxAccessLog.INDEX_NAME)
                    .source(JSON.toJSONString(nginxAccessLog), XContentType.JSON);
            IndexResponse response = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
            log.info("创建状态：{}", response.status());
        } catch (IOException e) {
            log.info("创建失败：{}", e.getMessage(), e);
        }
    }

}